package kafka

import (
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/gogf/gf/frame/g"
	"github.com/pkg/errors"
	"io/ioutil"
	"time"
)

const errorQueueSize = 32

// apply env config properties to a Sarama consumer config
func configureConsumer(saramaConf *sarama.Config) {
	saramaConf.ClientID = "kafka-consumer"
	saramaConf.Consumer.Return.Errors = true
	saramaConf.Consumer.Offsets.CommitInterval = 10 * time.Second
	saramaConf.Consumer.Group.Rebalance.Timeout = 1 * time.Minute
	saramaConf.Consumer.Group.Rebalance.Retry.Max = 6
	saramaConf.Consumer.Group.Rebalance.Retry.Backoff = 2 * time.Second
	saramaConf.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
	saramaConf.Consumer.Offsets.Initial = sarama.OffsetOldest
	configureTLS(saramaConf)
}

// apply env config properties to a Sarama consumer config
func configurePartitionConsumer(saramaConf *sarama.Config) {
	saramaConf.ClientID = "kafka-partition-consumer"
	saramaConf.Consumer.Return.Errors = true
	saramaConf.Consumer.Offsets.CommitInterval = 10 * time.Second
	saramaConf.Consumer.Group.Rebalance.Timeout = 1 * time.Minute
	saramaConf.Consumer.Group.Rebalance.Retry.Max = 6
	saramaConf.Consumer.Group.Rebalance.Retry.Backoff = 2 * time.Second
	saramaConf.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
	saramaConf.Consumer.Offsets.Initial = sarama.OffsetNewest
	configureTLS(saramaConf)
}

// apply env config properties into a Sarama producer config
func configureSyncProducer(saramaConf *sarama.Config) {
	saramaConf.ClientID = "kafka-sync-producer"
	saramaConf.Producer.RequiredAcks = sarama.WaitForLocal     // Only wait for the leader to ack
	saramaConf.Producer.Compression = sarama.CompressionSnappy // Compress messages
	saramaConf.Producer.Flush.Frequency = 1 * time.Second
	saramaConf.Producer.Flush.Frequency = 1 * time.Second
	saramaConf.Producer.Return.Successes = true
	saramaConf.Producer.Return.Errors = true
	configureTLS(saramaConf)
}

// apply env config properties into a Sarama producer config
func configureAsyncProducer(saramaConf *sarama.Config) {
	saramaConf.ClientID = "kafka-async-producer"
	saramaConf.Producer.RequiredAcks = sarama.WaitForLocal     // Only wait for the leader to ack
	saramaConf.Producer.Compression = sarama.CompressionSnappy // Compress messages
	saramaConf.Producer.Flush.Frequency = 1 * time.Second
	saramaConf.Producer.Flush.Frequency = 1 * time.Second
	saramaConf.Producer.Return.Successes = true
	saramaConf.Producer.Return.Errors = true
	configureTLS(saramaConf)
}

func configureSasl(saramaConf *sarama.Config) {
	config := g.Cfg().GetMapStrStr("kafka.sasl")

	// SASL
	if config["is_enable"] == "true" {
		saramaConf.Net.SASL.Enable = true
		// user = <消费组的账号>-<消费组ID>
		saramaConf.Net.SASL.User = fmt.Sprintf("%s-%s", config["user"], config["group"])
		saramaConf.Net.SASL.Password = config["password"]
	}
}

// side effect TLS setup into Sarama config if env config specifies to do so
func configureTLS(saramaConf *sarama.Config) error {
	config := g.Cfg().GetMapStrStr("kafka.tls")

	// configure TLS
	if config["is_enable"] == "true" {
		cert, err := tls.LoadX509KeyPair(config["cert"], config["key"])
		if err != nil {
			return errors.Wrapf(err, "failed to load TLS cert(%s) and key(%s)", config["cert"], config["key"])
		}

		ca, err := ioutil.ReadFile(config["cacerts"])
		if err != nil {
			return errors.Wrapf(err, "failed to load CA cert bundle at: %s", config["cacerts"])
		}

		pool := x509.NewCertPool()
		pool.AppendCertsFromPEM(ca)

		tlsCfg := &tls.Config{
			Certificates: []tls.Certificate{cert},
			RootCAs:      pool,
		}

		saramaConf.Net.TLS.Enable = true
		saramaConf.Net.TLS.Config = tlsCfg
	}

	return nil
}
